package com.my.code.guava;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.*;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * guava ListenableFuture 测试
 * guava version 29.0-jre
 * ListenableFuture 中的基础方法是addListener(Runnable, Executor)
 * 该方法会在多线程运算完的时候，指定的Runnable会被指定的Executor执行。<br/>
 * Futures.addCallback 会在任务完成后执行callback
 */
@SuppressWarnings("all")
public class ListenableFutureDemo {

    private static ListeningExecutorService service = service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(4));

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ListenableFutureDemo demo = new ListenableFutureDemo();

        ListenableFuture future1 = service.submit(() -> {
            int wait = 3;
            TimeUnit.SECONDS.sleep(wait);
            System.out.println("task1 is runned after " + wait + "s");
            return 1111;
        });
        ListenableFuture future2 = service.submit(() -> {
            int wait = 1;
            TimeUnit.SECONDS.sleep(wait);
            System.out.println("task2 is runned after " + wait + "s");
            return "2222";
        });
        ListenableFuture future3 = service.submit(() -> {
            int wait = 4;
            try {
                TimeUnit.SECONDS.sleep(wait);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task3 is runned after " + wait + "s");
            return "33333";//取消或异常，此结果在callback中不返回
        });

        Iterable futures = Lists.newArrayList(future1, future2, future3);
        //单独为future1添加结束监听
        future1.addListener(() -> {
            try {
                System.out.println("future1.listener report result=" + future1.get());
                future3.cancel(true);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }, service);

        demo.listenFutures(futures);
        System.out.println("main thread is end");
    }

    /**
     * 多个future callback
     */
    public void listenFutures(final Iterable futures) throws InterruptedException, ExecutionException {
        /**
         * Futures.allAsList(futures) 如果有线程取消或者计算失败，则Callback进入onFailure方法
         * Futures.successfulAsList(futures) 忽略取消或失败的线程，Callback进onSuccess方法，失败的线程结果以null代替
         */
        ListenableFuture all = Futures.allAsList(futures);
        Futures.addCallback(all, new FutureCallback() {
            @Override
            public void onSuccess(Object result) {//结果数组按照传入顺序
                System.out.println("--on success callback results:" + result);
                if (!service.isShutdown()) {
                    service.shutdownNow();
                }
            }

            @Override
            public void onFailure(Throwable t) {
                System.out.println("--all tasks finished, has failure");
                t.printStackTrace();
                service.shutdownNow();
            }
        }, service);
    }

    /**
     * 单个future callback
     */
    public void listenFuture(final ListenableFuture future) {
        Futures.addCallback(future, new FutureCallback() {
            @Override
            public void onSuccess(Object result) {
                System.out.println("task finish success");
            }

            @Override
            public void onFailure(Throwable t) {
                System.out.println("task finish failure");
            }
        }, service);
    }

}
